package com.talkingdata.hadoop.guard.service;

import com.talkingdata.hadoop.guard.bean.*;
import com.talkingdata.hadoop.guard.mapper.DfsClusterMapper;
import com.talkingdata.hadoop.guard.mapper.NamenodeMapper;
import com.talkingdata.hadoop.guard.mapper.RmMapper;
import com.talkingdata.hadoop.guard.mapper.YarnClusterMapper;
import com.talkingdata.hadoop.guard.util.LoadConfig;
import org.apache.hadoop.security.UserGroupInformation;
import org.apache.log4j.Logger;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import org.thymeleaf.TemplateEngine;
import org.thymeleaf.context.Context;

import java.io.IOException;
import java.util.*;


/**
 * 集群巡检日报:
 * 1. hdfs 集群状态
 * 2. hdfs 管理节点状态
 * 3. yarn 集群状态
 * 4. yarn 管理节点状态
 */
@Service
public class DailyReport {
    public static Logger log = Logger.getLogger(DailyReport.class);

    @Value("${spring.mail.username}")
    private String sender;

    @Value("${mail.subject}")
    private String subject;

    @Value("#{'${mail.recipients}'.split(',')}")
    private String[] recipients;

    // 通过 Context 构造模版中变量需要的值
    Context ctx = new Context();

    @Autowired
    private HadoopHandlerSerivce hadoopHandlerSerivce;

    @Autowired
    private MailService mailService;

    @Autowired
    private TemplateEngine templateEngine;

    @Autowired
    private OtherHadoopVersion otherHadoopVersion;

    @Autowired
    private DfsClusterMapper dfsClusterMapper;

    @Autowired
    private RmMapper rmMapper;

    @Autowired
    private NamenodeMapper namenodeMapper;

    @Autowired
    private YarnClusterMapper yarnClusterMapper;

    @Autowired
    LoadConfig loadConfig;

    public void HadoopReport() throws IOException {
        ArrayList<DfsClusterState> hdfsStateList = new ArrayList<>();
        ArrayList<YarnClusterState> yarnStatusList = new ArrayList<>();
        List<NameNodeState> nnlist = new ArrayList<>();
        ArrayList<RmState> rmList = new ArrayList<>();
        ArrayList<XmlConfig> configure = loadConfig.readxml();
        for (XmlConfig xc : configure) {
            if ("3.1.0".equals(xc.getVersion())) {
                hadoopHandlerSerivce.dfsInit(xc);
                // 获取集群namenode 管理节点信息 ， 找到activenamenode 节点
                nnlist.addAll(hadoopHandlerSerivce.getNamenodeState());
                // 获取集群状态信息
                DfsClusterState hdfsState = hadoopHandlerSerivce.getDfsClusterStatus();
                hdfsStateList.add(hdfsState);
                // 初始化 yarn集群
                hadoopHandlerSerivce.yarnInit(xc);
                rmList.addAll(hadoopHandlerSerivce.getRMState());
                // 获取 Yarn 集群信息
                YarnClusterState yarnStatus = hadoopHandlerSerivce.getYarnClusterStatus();
                yarnStatusList.add(yarnStatus);
                // 获取集群队列信息
                ArrayList<SchedulerState> schedulerList = hadoopHandlerSerivce.getScheduler();

            } else if ("2.6.0-cdh5.4.5".equals(xc.getVersion())) {
                if ("bjxgapp".equals(xc.getServiceId())) {
                    // 获取 hdfs 集群信息
                    DfsClusterState dfsCluster = dfsClusterMapper.selectDfsClusterStateByServiceId(xc.getServiceId());
                    System.out.println(dfsCluster);
                    hdfsStateList.add(dfsCluster);
                    // 获取集群namenode 节点信息
                    ArrayList<String[]> nnidList = otherHadoopVersion.getnnId(xc);
                    for (String[] s : nnidList) {
                        System.out.println(s[0] +"="+s[1]);
                        NameNodeState nameNodeState = namenodeMapper.selectNameNodeStatebynnId(s[0], s[1]);
                        System.out.println(nameNodeState.toString());
                        nnlist.add(nameNodeState);
                    }
                    // 获取 集群 resourcemanager 节点信息
                    RmState rmState = rmMapper.selectRmStatebyRmid("bjxgappyarn01", "single");
                    System.out.println(rmState.toString());
                    rmList.add(rmState);
                    // 获取 yarn集群信息
                    YarnClusterState yarnClusterState = yarnClusterMapper.selectYarnClusterStateByid("bjxgappyarn01");
                    System.out.println(yarnClusterState.toString());
                    yarnStatusList.add(yarnClusterState);
                } else {
                    // 获取 hdfs 集群信息
                    DfsClusterState dfsCluster = dfsClusterMapper.selectDfsClusterStateByServiceId(xc.getServiceId());
                    hdfsStateList.add(dfsCluster);
                    // 获取集群namenode 节点信息
                    ArrayList<String[]> nnidList = otherHadoopVersion.getnnId(xc);
                    for (String[] s : nnidList) {
                        //System.out.println(s[0] +"="+s[1]);
                        NameNodeState nameNodeState = namenodeMapper.selectNameNodeStatebynnId(s[0], s[1]);
                        nnlist.add(nameNodeState);
                    }
                    // 获取 集群 resourcemanager 节点信息
                    ArrayList<String[]> rmidList = otherHadoopVersion.getRmId(xc);
                    for (String[] s : rmidList) {
                        System.out.println(s[0] + "=" + s[1]);
                        RmState rmState = rmMapper.selectRmStatebyRmid(s[0], s[1]);
                        rmList.add(rmState);
                    }
                    // 获取 yarn集群信息
                    String yarnClusterId = otherHadoopVersion.getYarnClusterId(xc);
                    YarnClusterState yarnClusterState = yarnClusterMapper.selectYarnClusterStateByid(yarnClusterId);
                    System.out.println(yarnClusterState.toString());
                    yarnStatusList.add(yarnClusterState);
                }
            } else if ("2.7.3".equals(xc.getVersion())) {
                if ("bjdcnameservices01".equals(xc.getServiceId())) {
                    // 获取 hdfs 集群信息
                    DfsClusterState dfsCluster = dfsClusterMapper.selectDfsClusterStateByServiceId(xc.getServiceId());
                    System.out.println(dfsCluster);
                    hdfsStateList.add(dfsCluster);
                    // 获取集群namenode 节点信息
                    ArrayList<String[]> nnidList = otherHadoopVersion.getnnId(xc);
                    for (String[] s : nnidList) {
                        System.out.println(s[0] + "=" + s[1]);
                        NameNodeState nameNodeState = namenodeMapper.selectNameNodeStatebynnId(s[0], s[1]);
                        System.out.println(nameNodeState.toString());
                        nnlist.add(nameNodeState);
                    }
                    // 获取 集群 resourcemanager 节点信息
                    RmState rmState = rmMapper.selectRmStatebyRmid("bjdchbase01yarn01", "single");
                    System.out.println(rmState.toString());
                    rmList.add(rmState);
                    // 获取 yarn集群信息
                    YarnClusterState yarnClusterState = yarnClusterMapper.selectYarnClusterStateByid("bjdchbase01yarn01");
                    System.out.println(yarnClusterState.toString());
                    yarnStatusList.add(yarnClusterState);
                }
            }
            // 清空 kerberos 认证
            UserGroupInformation.reset();
        }
        ctx.setVariable("hdfsStateList", hdfsStateList);
        ctx.setVariable("yarnStatusList", yarnStatusList);
        ctx.setVariable("nnlist", nnlist);
        ctx.setVariable("rmList", rmList);
        log.info("发送邮件");
        // 使用TemplateEngine 对模版进行渲染
        String mail = templateEngine.process("dfsclusterstate.html", ctx);
        // 将渲染结果通过邮件发出
        mailService.sendHtmlMail(sender, recipients, subject, mail);
        log.info("邮件发送结束");
    }
}
